package com.easy.dev.init;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.easy.dev.entity.BeanConfig;
import com.easy.dev.factory.KafkaSourceFactory;
import com.easy.dev.mapper.BeanConfigMapper;
import com.easy.dev.utils.SpringBeanUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
@Component
public class KafkaSourceFactoryInit implements CommandLineRunner {

    @Autowired
    private BeanConfigMapper beanConfigMapper;

    @Autowired
    private KafkaSourceFactory kafkaSourceFactory;

    @Autowired
    private SpringBeanUtils springBean;

    @Override
    public void run(String... args) throws Exception {
        log.info("开始读取kafka配置");
        QueryWrapper wrapper = new QueryWrapper();
        wrapper.eq("type", "kafka");
        wrapper.orderByAsc("bean_name", "id");
        List<BeanConfig> beanConfigList = beanConfigMapper.selectList(wrapper);
        Map<String, List<BeanConfig>> map = new HashMap<>();
        beanConfigList.forEach(beanConfig -> {
            String name = beanConfig.getBeanName();
            List<BeanConfig> list = map.get(name);
            if (list == null) {
                list = new ArrayList<>();
                map.put(name, list);
            }
            list.add(beanConfig);
        });
        map.forEach((key, list) -> {
            KafkaListenerContainerFactory factory = kafkaSourceFactory.kafkaListenerContainerFactory(list);
            springBean.registerBean(key, factory);
        });
    }
}
